package com.shujia.custom;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;

/*
    udtf编程
 */
public class UDTFDemo1 extends GenericUDTF {

    /*
        initialize方法中我们要定义输出的类型和字段值名

        举例：M1001#xiaohu#S324231212,lg#M1002#S2543412432,S21312312412#M1003#qzq

        输出：
            1001    xiaohu  324231212
            1002    lg  2543412432
            1003    qzq 21312312412
     */
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        //创建一个List集合存储结果列的名字
        ArrayList<String> colNames = new ArrayList<>();
        //创建一个集合存储每一列的数据类型
        ArrayList<ObjectInspector> colTypes = new ArrayList<>();

        //向集合中添加元素，设置列的名字以及类型
        colNames.add("id");
        colTypes.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

        colNames.add("name");
        colTypes.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

        colNames.add("cardId");
        colTypes.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

        //返回一个对象，该对象封装了列的名字以及列的类型
        return ObjectInspectorFactory.getStandardStructObjectInspector(colNames, colTypes);
    }

    /*
        最重要的实现方法，将输入进来的数据，按照initialize方法的格式转换输出
        args: 接收将来函数调用时传入的参数

        将来需要使用forward函数将每一行输出，这里需要将输出一行的每一列数据封装成一个数组
     */
    @Override
    public void process(Object[] args) throws HiveException {
        //M1001#xiaohu#S324231212,lg#M1002#S2543412432,S21312312412#M1003#qzq
        //select lgFunc("M1001#xiaohu#S324231212,lg#M1002#S2543412432,S21312312412#M1003#qzq") from xxx;
        String info = args[0].toString();

        //定义一个数组
        String[] cols = new String[3];

        String[] students = info.split(",");
        for (String student : students) {
            //M1001#xiaohu#S324231212
            String[] infos = student.split("#");
            for (String i : infos) {
                if(i.startsWith("M")){
                    cols[0] = i.substring(1);
                }else if(i.startsWith("S")){
                    cols[2] = i.substring(1);
                }else {
                    cols[1] = i;
                }
            }
            forward(cols); // 将一行的数据输出出去
        }
    }

    @Override
    public void close() throws HiveException {
        //释放initialize方法中创建的资源
    }
}
